package net.bwie.realtime.jtp.dws.douyin.log.functions;

import net.bwie.realtime.jtp.dws.douyin.log.bean.*;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.*;

public class DouYinOrderWindowFunction5 implements WindowFunction<
        UnifiedData5,                // 输入：统一数据载体
        GoodsStatMetric5,            // 输出：商品指标
        Long,                       // 分组键：商品ID（skuId）
        TimeWindow> {               // 时间窗口

    @Override
    public void apply(
            Long skuId,                  // 分组键：商品ID
            TimeWindow window,           // 窗口信息
            Iterable<UnifiedData5> input, // 窗口内所有数据
            Collector<GoodsStatMetric5> out) throws Exception {

        // 1. 初始化统计容器（与OrderSourceRatioWindowFunction3结构一致）
        long clickCount = 0L;                  // 商品点击次数（日志表）
        Set<Long> totalUserSet = new HashSet<>(); // 总人数（去重用户ID，两张表合并）
        long dealOrderCount = 0L;              // 成交订单数（订单表，finished状态）
        double dealAmount = 0.0;               // 成交金额（double累加）

        // 2. 遍历窗口数据，按类型统计（通过UnifiedData的flag区分数据来源）
        Iterator<UnifiedData5> iterator = input.iterator();
        while (iterator.hasNext()) {
            UnifiedData5 unifiedData = iterator.next();
            Integer flag = unifiedData.getEventTypeFlag();

            // 2.1 处理日志表的“点击事件”（flag=1，logData有效）
            if (flag == 1 && unifiedData.getLogData() != null) {
                EventLog6 clickEvent = unifiedData.getLogData();
                clickCount++; // 累加点击次数
                if (clickEvent.getUserId() != null) {
                    totalUserSet.add(clickEvent.getUserId()); // 记录点击用户
                }
            }

            // 2.2 处理订单表的“成交事件”（flag=2，orderData有效）
            else if (flag == 2 && unifiedData.getOrderData() != null) {
                OrderDetail5 order = unifiedData.getOrderData();
                if ("canceled".equals(order.getStatus())) { // 仅统计已完成订单
                    dealOrderCount++; // 累加成交订单数
                    dealAmount += order.getOrderAmount(); // 累加成交金额（double）
                }
                if (order.getUserId() != null) {
                    totalUserSet.add(order.getUserId()); // 记录下单用户
                }
            }
        }

        // 3. 计算核心指标（避免除零错误，参照原类占比计算逻辑）
        long totalUserCount = totalUserSet.size();
        // 点击率 = 点击次数 / 总人数（保留4位小数）
        double goodsClickRate = totalUserCount == 0 ? 0.0 :
                Math.round((clickCount * 1.0 / totalUserCount) * 10000) / 10000.0;
        // 转化率 = 成交订单数 / 总人数（保留4位小数）
        double goodsConversionRate = totalUserCount == 0 ? 0.0 :
                Math.round((dealOrderCount * 1.0 / totalUserCount) * 10000) / 10000.0;

        // 4. 格式化窗口时间（与原类完全一致，使用DateTimeUtil）
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), "yyyy-MM-dd HH:mm:ss");
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), "yyyy-MM-dd HH:mm:ss");
        String curDate = windowStartTime.substring(0, 10); // 提取日期

        // 5. 封装指标（商品名称/价格预留维表关联位）
        GoodsStatMetric5 metric = new GoodsStatMetric5(
                windowStartTime,
                windowEndTime,
                curDate,
                skuId,
                "默认商品名", // 实际需关联商品维表
                99.99,      // 实际需关联商品维表（double）
                goodsClickRate,
                dealOrderCount,
                dealAmount,
                goodsConversionRate
        );
        out.collect(metric);
    }
}
